package com.atguigu.bigdata.edu.realtime.app.dwd;


import com.atguigu.bigdata.edu.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class Dwd_02_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradeCartAdd().init(
                3020,
                2,
                "Dwd_02_DwdTradeCartAdd"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 消费 ods_db
        // ddl方式
        readOdsDb(tEnv, "Dwd_02_DwdTradeCartAdd");
        // sql中的 print 是阻塞式方法
        //        tEnv.sqlQuery("select * from ods_db").execute().print();
        // 2. 从 ods_db中过滤出加购
        Table cartInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['course_id'] course_id, " +
                " `data`['course_name'] course_name, " +
                " `data`['session_id'] session_id, " +
                " `data`['create_time'] create_time, " +
                " `data`['update_time'] update_time, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='cart_info' " +
                "and (" +
                " `type`='insert' " +
                "  or " +
                "   (`type`='update' " +
                "     and `old`['sold'] is not null " +
                "     and cast(`data`['sold'] as int) > cast(`old`['sold']  as int)" +
                "   )" +
                ")");
        //cartInfo.execute().print();

        // 5. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                "id string,  " +
                "user_id string,  " +
                "course_id string,  " +
                "course_name string, " +
                "session_id string, " +
                "create_time string, " +
                "update_time string, " +
                "ts bigint  " +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD));

        cartInfo.executeInsert("dwd_trade_cart_add");
    }
}
/*
加购的过滤:
    database = gmall2022
    table = cart_info
    type= insert update
    
    sku_1      2     新增
    sku_1      5-2   update
        insert 或者 update 的时候,sku_num 变大
 */